package com.itcode;


import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.calcite.CalciteConfig;
import org.apache.flink.types.Row;
import org.apache.hadoop.security.UserGroupInformation;

import java.io.IOException;

/**
 * Flink CDC的样例类
 *
 * @author code1997
 */
public class FlinkCdcBySql {

    public static final String ENV = "env";
    public static final String LOCAL_ENV = "local";


    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(executionEnvironment);
        tableEnvironment.getConfig().setPlannerConfig(CalciteConfig.DEFAULT());
        String sourceDdl = "create table mysql_source (" +
                "db_name STRING METADATA FROM 'database_name' VIRTUAL," +
                "table_name STRING METADATA  FROM 'table_name' VIRTUAL," +
                "id INT," +
                "activity_name String," +
                "PRIMARY KEY(id) NOT ENFORCED" +
                ") WITH (" +
                "'connector'='mysql-cdc'," +
                "'hostname'='hadoop02'," +
                "'port'='3307'," +
                "'username'='root'," +
                "'password'='19971001'," +
                "'database-name'='gmall_flink'," +
                "'table-name'='activity_info')";
        tableEnvironment.executeSql(sourceDdl);
        String query = "select * from mysql_source";
        Table result = tableEnvironment.sqlQuery(query);
        result.printSchema();
        //将table转换成流进行打印
        tableEnvironment.toRetractStream(result, Row.class).print();
        executionEnvironment.execute();

    }
}
